package io.kiki.sba.registry.server.data_store.slot;

import io.kiki.sba.registry.common.model.Tuple;
import io.kiki.sba.registry.common.model.dataserver.DatumSummary;
import io.kiki.sba.registry.common.model.dataserver.DatumVersion;
import io.kiki.sba.registry.common.model.slot.*;
import io.kiki.sba.registry.common.model.slot.filter.SyncSlotAcceptorManager;
import io.kiki.sba.registry.common.model.slot.func.SlotFunction;
import io.kiki.sba.registry.common.model.slot.func.SlotFunctionRegistry;
import io.kiki.sba.registry.server.data_store.bootstrap.ServerConfig;
import io.kiki.sba.registry.server.data_store.cache.DatumStorageDelegate;
import io.kiki.sba.registry.server.data_store.change.DataChangeEventCenter;
import io.kiki.sba.registry.server.data_store.change.DataChangeType;
import io.kiki.sba.registry.server.data_store.lease.ClientInterfaceLeaseManager;
import io.kiki.sba.registry.server.data_store.pubiterator.DatumBiConsumer;
import io.kiki.sba.registry.server.data_store.remoting.DataNodeExchanger;
import io.kiki.sba.registry.server.data_store.remoting.SessionNodeExchanger;
import io.kiki.sba.registry.server.data_store.remoting.metaserver.MetaServerServiceImpl;
import io.kiki.sba.registry.server.shared.env.ServerEnv;
import io.kiki.sba.registry.server.shared.slot.SlotsRecorder;
import io.kiki.sba.registry.task.KeyedTask;
import io.kiki.sba.registry.task.KeyedThreadPoolExecutor;
import io.kiki.sba.registry.task.TaskErrorSilenceException;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.StringUtil;
import io.kiki.sba.registry.util.WakeUpLoopExecuteTask;
import lombok.Getter;
import lombok.Setter;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

import static io.kiki.sba.registry.server.data_store.slot.SlotMetrics.Manager.*;

@Setter
@Getter
public final class SlotManagerImpl implements SlotManager {
    private static final Logger logger = LoggerFactory.getLogger(SlotManagerImpl.class);


    private final SlotFunction slotFunction = SlotFunctionRegistry.getSlotFunction();
    /**
     * the sync and migrating may happen parallelly when slot role has modified. make sure the datum
     * merging is idempotent
     */
    private final SyncingWatchDog watchDog = new SyncingWatchDog();
    private final AtomicReference<Slots> updatingSlots = new AtomicReference<Slots>();
    private final ReadWriteLock updateLock = new ReentrantReadWriteLock();
    private final SlotsStates SlotsStates = new SlotsStates();
    @Autowired
    private DataNodeExchanger dataNodeExchanger;
    @Autowired
    private SessionNodeExchanger sessionNodeExchanger;
    @Autowired
    private MetaServerServiceImpl metaServerService;
    @Autowired
    private ServerConfig serverConfig;
    @Resource
    private DatumStorageDelegate datumStorageDelegate;
    @Autowired
    private DataChangeEventCenter dataChangeEventCenter;
    @Autowired
    private ClientInterfaceLeaseManager clientInterfaceLeaseManager;
    @Autowired(required = false)
    private List<SlotsRecorder> recorders;
    @Resource
    private SyncSlotAcceptorManager syncSlotAcceptAllManager;
    @Autowired
    private SlotChangeListenerManager slotChangeListenerManager;
    private KeyedThreadPoolExecutor migrateSessionExecutor;
    private KeyedThreadPoolExecutor syncSessionExecutor;
    private KeyedThreadPoolExecutor syncLeaderExecutor;

    private static SlotRole getRole(Slot s) {
        return localIsLeader(s) ? SlotRole.leader : SlotRole.follower;
    }

    private static boolean localIsLeader(Slot slot) {
        return ServerEnv.isLocalServer(slot.getLeaderDataStoreNodeId());
    }

    @PostConstruct
    public void init() {
        initExecutors();
        ConcurrentUtils.createDaemonThread("SyncingWatchDog", watchDog).start();
    }

    void initExecutors() {
        this.migrateSessionExecutor = new KeyedThreadPoolExecutor("migrate-session", serverConfig.getSlotLeaderSyncSessionExecutorThreadSize(), serverConfig.getSlotLeaderSyncSessionExecutorQueueSize());

        this.syncSessionExecutor = new KeyedThreadPoolExecutor("sync-session", serverConfig.getSlotLeaderSyncSessionExecutorThreadSize(), serverConfig.getSlotLeaderSyncSessionExecutorQueueSize());

        this.syncLeaderExecutor = new KeyedThreadPoolExecutor("sync-leader", serverConfig.getSlotFollowerSyncLeaderExecutorThreadSize(), serverConfig.getSlotFollowerSyncLeaderExecutorQueueSize());
    }

    @Override
    public int slotOf(String dataInfoId) {
        return slotFunction.slotOf(dataInfoId);
    }

    @Override
    public Slot getSlot(String dataCenter, int slotId) {
        final SlotState slotState = SlotsStates.slotStates.get(slotId);
        return slotState == null ? null : slotState.slot;
    }

    @Override
    public SlotAccess checkSlotAccess(String dataCenter, int slotId, long srcSlotEpoch, long srcLeaderEpoch) {
        Slots currentSlots;
        SlotState slotState;
        updateLock.readLock().lock();
        try {
            currentSlots = SlotsStates.slots;
            slotState = SlotsStates.slotStates.get(slotId);
        } finally {
            updateLock.readLock().unlock();
        }

        final long currentEpoch = currentSlots.getEpoch();
        if (currentEpoch < srcSlotEpoch) {
            triggerUpdateSlots(srcSlotEpoch);
        }
        return checkSlotAccess(slotId, currentEpoch, slotState, srcLeaderEpoch);
    }

    public SlotAccess checkSlotAccess(int slotId, long currentslotsEpoch, ISlotState slotState, long srcLeaderEpoch) {
        if (slotState == null) {
            return new SlotAccess(slotId, currentslotsEpoch, SlotAccess.Status.moved, -1);
        }
        final Slot slot = slotState.getSlot();
        if (!localIsLeader(slot)) {
            return new SlotAccess(slotId, currentslotsEpoch, SlotAccess.Status.moved, slot.getLeaderEpoch());
        }
        if (!slotState.isMigrated()) {
            return new SlotAccess(slotId, currentslotsEpoch, SlotAccess.Status.migrating, slot.getLeaderEpoch());
        }
        if (slot.getLeaderEpoch() != srcLeaderEpoch) {
            return new SlotAccess(slotId, currentslotsEpoch, SlotAccess.Status.mismatch, slot.getLeaderEpoch());
        }
        return new SlotAccess(slotId, currentslotsEpoch, SlotAccess.Status.accept, slot.getLeaderEpoch());
    }

    public boolean hasSlot() {
        updateLock.readLock().lock();
        try {
            for (SlotState slotState : SlotsStates.slotStates.values()) {
                Slot slot = slotState.slot;
                if (StringUtils.equals(slot.getLeaderDataStoreNodeId(), ServerEnv.IP)) {
                    return true;
                }
                if (slot.getFollowerDataStoreNodeIdSet() != null && slot.getFollowerDataStoreNodeIdSet().contains(ServerEnv.IP)) {
                    return true;
                }
            }
        } finally {
            updateLock.readLock().unlock();
        }
        return false;
    }

    @Override
    public List<BaseSlotStatus> getSlotStatuses() {
        List<BaseSlotStatus> slotStatuses = new ArrayList<>(SlotsStates.slotStates.size());
        updateLock.readLock().lock();
        try {
            for (Map.Entry<Integer, SlotState> entry : SlotsStates.slotStates.entrySet()) {
                int slotId = entry.getKey();
                SlotState slotState = entry.getValue();
                if (localIsLeader(slotState.slot)) {
                    LeaderSlotStatus status = new LeaderSlotStatus(slotId, slotState.slot.getLeaderEpoch(), ServerEnv.IP, slotState.migrated ? BaseSlotStatus.LeaderStatus.healthy : BaseSlotStatus.LeaderStatus.unhealthy);
                    slotStatuses.add(status);
                } else {
                    final KeyedTask syncLeaderTask = slotState.syncLeaderTask;
                    FollowerSlotStatus status = new FollowerSlotStatus(slotId, slotState.slot.getLeaderEpoch(), ServerEnv.IP, syncLeaderTask != null ? syncLeaderTask.getStartTime() : 0, slotState.lastSuccessLeaderSyncTime);
                    slotStatuses.add(status);
                }
            }
            return slotStatuses;
        } finally {
            updateLock.readLock().unlock();
        }
    }

    @Override
    public boolean isLeader(String dataCenter, int slotId) {
        final SlotState slotState = SlotsStates.slotStates.get(slotId);
        return slotState != null && localIsLeader(slotState.slot);
    }

    @Override
    public boolean isFollower(String dataCenter, int slotId) {
        final SlotState slotState = SlotsStates.slotStates.get(slotId);
        return slotState != null && slotState.slot.getFollowerDataStoreNodeIdSet().contains(ServerEnv.IP);
    }

    @Override
    public boolean updateSlots(Slots update) {
        final Slots curSlots = this.SlotsStates.slots;
        if (curSlots.getEpoch() >= update.getEpoch()) {
            return false;
        }
        final Slots updating = this.updatingSlots.get();
        if (updating != null && updating.getEpoch() >= update.getEpoch()) {
            return false;
        }
        recordSlots(update);
        // confirmed that Slots is related to us
        update = update.filter(ServerEnv.IP);

        curSlots.assertSlotLessThan(update);
        if (updating != null) {
            updating.assertSlotLessThan(update);
        }

        // do that async, not block the heartbeat
        // v2 > v1, compareAndSet to avoid v1 cover v2
        if (updatingSlots.compareAndSet(updating, update)) {
            watchDog.wakeup();
            logger.info("updating slot table, new={}, current={}", update.getEpoch(), curSlots.getEpoch());
            return true;
        }
        return false;
    }

    private void recordSlots(Slots slots) {
        if (recorders == null) {
            return;
        }
        for (SlotsRecorder recorder : recorders) {
            recorder.record(slots);
        }
    }

    private void updateSlotState(Slots slots) {
        for (Slot slot : slots.getSlotIdToSlotMap()) {
            SlotState slotState = SlotsStates.slotStates.get(slot.getId());
            listenAddUpdate(slot);
            if (slotState != null) {
                slotState.update(slot);
            } else {
                SlotsStates.slotStates.put(slot.getId(), new SlotState(slot));
                logger.info("add slot, slot={}", slot);
            }
        }

        final Iterator<Map.Entry<Integer, SlotState>> iterator = SlotsStates.slotStates.entrySet().iterator();
        while (iterator.hasNext()) {
            Map.Entry<Integer, SlotState> entry = iterator.next();
            if (slots.getSlot(entry.getKey()) == null) {
                final Slot slot = entry.getValue().slot;
                iterator.remove();
                // important, first remove the slot for GetData Access check, then clean the data
                listenRemoveUpdate(slot);
                observeLeaderMigratingFinish(slot.getId());
                logger.info("remove slot, slot={}", slot);
            }
        }
        SlotsStates.slots = slots;

        observeLeaderAssignGauge(SlotsStates.slots.getLeaderNum(ServerEnv.IP));
        observeFollowerAssignGauge(SlotsStates.slots.getFollowerNum(ServerEnv.IP));
    }

    public boolean processUpdating() {
        final Slots updating = updatingSlots.getAndSet(null);
        if (updating != null) {
            if (updating.getEpoch() > SlotsStates.slots.getEpoch()) {
                // lock for update, avoid the checkAccess get the wrong epoch
                updateLock.writeLock().lock();
                try {
                    updateSlotState(updating);
                } finally {
                    updateLock.writeLock().unlock();
                }
                List<DataStoreNode> leaders = updating.transfer(ServerEnv.IP, true);
                logger.info("updating slot table, leaders={}, {}, ", leaders, updating);
                return true;
            } else {
                logger.warn("skip updating={}, current={}", updating.getEpoch(), SlotsStates.slots.getEpoch());
            }
        }
        return false;
    }

    void syncWatch() {
        final int syncSessionIntervalMs = serverConfig.getSlotLeaderSyncSessionIntervalSecs() * 1000;
        final int syncLeaderIntervalMs = serverConfig.getSlotFollowerSyncLeaderIntervalSecs() * 1000;
        final long slotsEpoch = SlotsStates.slots.getEpoch();
        for (SlotState slotState : SlotsStates.slotStates.values()) {
            try {
                sync(slotState, syncSessionIntervalMs, syncLeaderIntervalMs, slotsEpoch);
            } catch (Throwable e) {
                logger.error("[syncCommit]failed to do sync slot {}, migrated={}", slotState.slot, slotState.migrated, e);
            }
        }
    }

    boolean sync(SlotState slotState, int syncSessionIntervalMs, int syncLeaderIntervalMs, long slotsEpoch) {
        final Slot slot = slotState.slot;
        if (localIsLeader(slot)) {
            final KeyedTask<SyncLeaderTask> syncLeaderTask = slotState.syncLeaderTask;
            if (syncLeaderTask != null && !syncLeaderTask.isFinished()) {
                // must wait the sync leader finish, avoid the sync-leader conflict with sync-session
                logger.warn("[waitSyncLeader]wait for sync-leader to finish, {},{}", slot, syncLeaderTask);
                return false;
            }
            slotState.syncLeaderTask = null;
            final Set<String> sessions = metaServerService.getSessionServerList();
            if (slotState.migrated) {

                syncSessions(slotState, sessions, syncSessionIntervalMs, slotsEpoch);
            } else {
                syncMigrating(slotState, sessions, syncSessionIntervalMs, slotsEpoch);
                // check all migrating task
                checkMigratingTask(slotState, sessions);
            }
        } else {
            // sync leader
            syncLeader(slotState, syncLeaderIntervalMs, slotsEpoch);
        }
        return true;
    }

    private boolean checkMigratingTask(SlotState slotState, Collection<String> sessions) {
        final Slot slot = slotState.slot;
        final long span = System.currentTimeMillis() - slotState.migratingStartTime;
        logger.info("[migrating]{},span={},tasks={}/{},sessions={}/{},remains={}", slotState.slotId, span, slotState.migratingTasks.size(), slotState.migratingTasks.keySet(), sessions.size(), sessions, getMigratingSpans(slotState));

        // monitor slow migrating
        if (span > 1000 * 8) {
            logger.error("[slowSlot]{},span={}", slotState.slotId, span);
        }

        // check all migrating task
        if (slotState.migratingTasks.isEmpty() || sessions.isEmpty()) {
            logger.warn("sessionNodes or migratingTask is empty when migrating, {}", slot);
            return false;
        }
        // TODO the session down and up in a short time. session.processId is important
        if (slotState.isAnywaySuccess(sessions)) {
            // after migrated, force to update the version
            // make sure the version is newly than old leader's
            Map<String, DatumVersion> versions = datumStorageDelegate.updateVersion(serverConfig.getDataCenter(), slotState.slotId);
            slotState.migrated = true;
            // versions has update, notify change
            dataChangeEventCenter.onChange(versions.keySet(), DataChangeType.migrated, serverConfig.getDataCenter());
            logger.info("[finish]slotId={}, span={}, slot={}, sessions={}", slotState.slotId, span, slot, sessions);
            slotState.migratingTasks.clear();
            observeLeaderMigratingFinish(slotState.slotId);
            observeLeaderMigratingHistogram(slotState.slotId, span);
            return true;
        }
        return false;
    }

    private Map<String, Long> getMigratingSpans(SlotState slotState) {
        final long now = System.currentTimeMillis();
        Map<String, Long> spans = new TreeMap<>();
        for (Map.Entry<String, MigratingTask> e : slotState.migratingTasks.entrySet()) {
            MigratingTask m = e.getValue();
            if (!m.task.isFinished() || (m.task.isFailed() && !m.forceSuccess)) {
                spans.put(e.getKey(), now - m.createTimestamp);
            }
        }
        return spans;
    }

    boolean triggerEmergencyMigrating(SlotState slotState, Collection<String> sessions, MigratingTask mtask, int notSyncedCount) {
        try {
            // session.size=1 means only one session, could not skip
            if (sessions.size() <= 1) {
                return false;
            }
            final long span = System.currentTimeMillis() - mtask.createTimestamp;
            int emergencyCases = 0;
            if (span > serverConfig.getMigratingMaxSecs() * 1000) {
                emergencyCases++;
            }
            if (mtask.tryCount > serverConfig.getMigratingMaxRetry()) {
                emergencyCases++;
            }
            if (emergencyCases != 0) {
                if (notSyncedCount <= serverConfig.getMigratingMaxUnavailable()) {
                    emergencyCases++;
                }
                logger.error("[slowSession]{},span={},try={},ing={}/{},session={}", slotState.slotId, span, mtask.tryCount, notSyncedCount, sessions.size(), mtask.sessionIp);
                if (emergencyCases >= 3) {
                    // mark the session migrating success, avoid block
                    mtask.forceSuccess = true;
                    return true;
                }
            }
        } catch (Throwable e) {
            // cache unexpect exception, make sure the check not break the migrating
            logger.error("failed to check slow migrating, slotId={}, session={}", slotState.slotId, mtask.sessionIp, e);
        }
        return false;
    }

    private void syncMigrating(SlotState slotState, Collection<String> sessions, int syncSessionIntervalMs, long slotsEpoch) {
        final Slot slot = slotState.slot;
        if (slotState.migratingStartTime == 0) {
            slotState.migratingStartTime = System.currentTimeMillis();
            slotState.migratingTasks.clear();
            observeLeaderMigratingStart(slot.getId());
            logger.info("start migrating, slotId={}, sessionSize={}, sessions={}", slotState.slotId, sessions.size(), sessions);
        }
        final int notSyncedCount = sessions.size() - slotState.countSyncSuccess(sessions);
        for (String sessionIp : sessions) {
            MigratingTask mtask = slotState.migratingTasks.get(sessionIp);
            if (mtask == null) {
                KeyedTask<SyncSessionTask> ktask = commitSyncSessionTask(slot, slotsEpoch, sessionIp, null, true);
                mtask = new MigratingTask(sessionIp, ktask);
                slotState.migratingTasks.put(sessionIp, mtask);
                logger.info("migrating start,slotId={},session={}", slot.getId(), sessionIp);
                continue;
            }

            if (mtask.task.isFailed() && !mtask.forceSuccess) {
                // failed and not force Success, try to trigger emergency
                if (triggerEmergencyMigrating(slotState, sessions, mtask, notSyncedCount)) {
                    logger.info("[emergency]{},session={}", slotState.slotId, mtask.sessionIp);
                } else {
                    KeyedTask<SyncSessionTask> ktask = commitSyncSessionTask(slot, slotsEpoch, sessionIp, null, true);
                    mtask.task = ktask;
                    mtask.tryCount++;
                    logger.error("migrating retry,slotId={},try={},session={},create={}/{}", slot.getId(), mtask.tryCount, sessionIp, mtask.createTimestamp, System.currentTimeMillis() - mtask.createTimestamp);
                    continue;
                }
            }
            // force success or migrating finish. try to sync session
            // avoid the time of migrating is too long and block the syncing of session
            if (mtask.task.isOverAfter(syncSessionIntervalMs)) {
                if (syncSession(slotState, sessionIp, null, syncSessionIntervalMs, slotsEpoch)) {
                    logger.info("slotId={}, sync session in migrating, session={}", slot.getId(), sessionIp);
                }
            }
        }
    }

    private void syncSessions(SlotState slotState, Collection<String> sessions, int syncSessionIntervalMs, long slotsEpoch) {

        final Set<String> doSyncSet = new HashSet<>(16);
        for (String sessionIp : sessions) {
            if (needSessionSync(slotState, sessionIp, syncSessionIntervalMs)) {
                doSyncSet.add(sessionIp);
            }
        }

        if (CollectionUtils.isEmpty(doSyncSet)) {
            return;
        }

        final Map<String, Map<String, DatumSummary>> datumSummary = new HashMap<>(doSyncSet.size());

        datumStorageDelegate.foreach(serverConfig.getDataCenter(), slotState.slotId, DatumBiConsumer.publisherGroupsBiConsumer(datumSummary, doSyncSet, syncSlotAcceptAllManager));
        for (String sessionIp : doSyncSet) {
            Map<String, DatumSummary> summary = datumSummary.get(sessionIp);
            syncSession(slotState, sessionIp, summary, syncSessionIntervalMs, slotsEpoch);
        }
    }

    /**
     * summary == null means can not assembly summary at first(migrating); do
     * getDatumSummary(sessionIp) later
     */
    private boolean syncSession(SlotState slotState, String sessionIp, Map<String, DatumSummary> summary, int syncSessionIntervalMs, long slotsEpoch) {
        final Slot slot = slotState.slot;
        if (needSessionSync(slotState, sessionIp, syncSessionIntervalMs)) {
            KeyedTask<SyncSessionTask> task = commitSyncSessionTask(slot, slotsEpoch, sessionIp, summary, false);
            slotState.syncSessionTasks.put(sessionIp, task);
            return true;
        }
        return false;
    }

    private boolean needSessionSync(SlotState slotState, String sessionIp, int syncSessionIntervalMs) {
        KeyedTask<SyncSessionTask> task = slotState.syncSessionTasks.get(sessionIp);
        return task == null || task.isOverAfter(syncSessionIntervalMs);
    }

    private void syncLeader(SlotState slotState, int syncLeaderIntervalMs, long slotsEpoch) {
        final Slot slot = slotState.slot;
        final KeyedTask<SyncLeaderTask> syncLeaderTask = slotState.syncLeaderTask;
        if (syncLeaderTask != null && syncLeaderTask.isFinished()) {
            slotState.completeSyncLeaderTask();
        }
        if (syncLeaderTask == null || syncLeaderTask.isOverAfter(syncLeaderIntervalMs)) {
            // sync leader no need to notify event
            SlotDiffSyncer syncer = new SlotDiffSyncer(serverConfig, datumStorageDelegate, null, clientInterfaceLeaseManager, syncSlotAcceptAllManager);
            SyncContinues continues = new SyncContinues() {
                @Override
                public boolean continues() {
                    return isFollower(serverConfig.getDataCenter(), slot.getId());
                }
            };
            SyncLeaderTask task = new SyncLeaderTask(serverConfig.getDataCenter(), serverConfig.getDataCenter(), slotsEpoch, slot, syncer, dataNodeExchanger, continues);
            slotState.syncLeaderTask = syncLeaderExecutor.execute(slot.getId(), task);
        } else
            if (!syncLeaderTask.isFinished()) {
                if (System.currentTimeMillis() - syncLeaderTask.getCreateTime() > 5000) {
                    // the sync leader is running more than 5secs, print
                    logger.info("sync-leader running, {}", syncLeaderTask);
                }
            }
    }

    private KeyedTask<SyncSessionTask> commitSyncSessionTask(Slot slot, long slotsEpoch, String sessionIp, Map<String, DatumSummary> summary, boolean migrate) {
        SlotDiffSyncer syncer = new SlotDiffSyncer(serverConfig, datumStorageDelegate, dataChangeEventCenter, clientInterfaceLeaseManager, syncSlotAcceptAllManager);
        SyncContinues continues = new SyncContinues() {
            @Override
            public boolean continues() {
                // if not leader, the syncing need to break
                return isLeader(serverConfig.getDataCenter(), slot.getId());
            }
        };
        SyncSessionTask task = new SyncSessionTask(serverConfig.getDataCenter(), migrate, slotsEpoch, slot, sessionIp, syncer, sessionNodeExchanger, continues, summary);
        if (migrate) {
            // group by slotId and session
            return migrateSessionExecutor.execute(new Tuple(slot.getId(), sessionIp), task);
        } else {
            // at most there is 4 tasks for a session, avoid too many tasks hit the same session
            return syncSessionExecutor.execute(new Tuple((slot.getId() % 4), sessionIp), task);
        }
    }

    @Override
    public void triggerUpdateSlots(long expectEpoch) {
        // TODO
    }

    @Override
    public Set<Integer> leaderSlotIds() {
        Set<Integer> ret = new TreeSet<>();
        updateLock.readLock().lock();
        try {
            for (SlotState slotState : SlotsStates.slotStates.values()) {
                Slot slot = slotState.slot;
                if (StringUtils.equals(slot.getLeaderDataStoreNodeId(), ServerEnv.IP)) {
                    ret.add(slotState.slotId);
                }
            }
        } finally {
            updateLock.readLock().unlock();
        }
        return ret;
    }

    @Override
    public Tuple<Long, List<BaseSlotStatus>> getslotsEpochAndStatuses(String dataCenter) {
        updateLock.readLock().lock();
        try {
            long slotsEpoch = getSlotsEpoch();
            List<BaseSlotStatus> slotStatuses = getSlotStatuses();
            return Tuple.of(slotsEpoch, slotStatuses);
        } finally {
            updateLock.readLock().unlock();
        }
    }

    @Override
    public long getSlotsEpoch() {
        return SlotsStates.slots.getEpoch();
    }

    private void listenAddUpdate(Slot s) {
        slotChangeListenerManager.localUpdateListeners().forEach(listener -> listener.onSlotAdd(serverConfig.getDataCenter(), s.getId(), getRole(s)));
    }

    private void listenRemoveUpdate(Slot s) {
        slotChangeListenerManager.localUpdateListeners().forEach(listener -> listener.onSlotRemove(serverConfig.getDataCenter(), s.getId(), getRole(s)));
    }


    public interface ISlotState {
        boolean isMigrated();

        Slot getSlot();
    }

    public static final class SlotsStates {
        final Map<Integer, SlotState> slotStates = new ConcurrentHashMap<>();
        volatile Slots slots = Slots.INIT;
    }

    static final class SlotState implements ISlotState {
        final int slotId;
        final Map<String, MigratingTask> migratingTasks = new TreeMap<>();
        final Map<String, KeyedTask<SyncSessionTask>> syncSessionTasks = new TreeMap<>();
        volatile Slot slot;
        volatile boolean migrated;
        volatile long migratingStartTime;
        volatile long lastSuccessLeaderSyncTime = -1L;
        volatile KeyedTask<SyncLeaderTask> syncLeaderTask;

        SlotState(Slot slot) {
            this.slotId = slot.getId();
            this.slot = slot;
        }

        void update(Slot s) {
            StringUtil.checkEquals(slotId, s.getId(), "slot.id");
            if (slot.getLeaderEpoch() != s.getLeaderEpoch()) {
                this.migrated = false;
                this.syncSessionTasks.clear();
                this.migratingTasks.clear();
                this.migratingStartTime = 0;
                if (localIsLeader(s)) {
                    // leader change
                    observeLeaderUpdateCounter();
                }
                observeLeaderMigratingFinish(slot.getId());
                logger.info("update slot with leaderEpoch, exist={}, now={}", slot, s);
            }
            this.slot = s;
            logger.info("update slot, slot={}", slot);
        }

        void completeSyncLeaderTask() {
            if (syncLeaderTask != null && syncLeaderTask.isSuccess()) {
                this.lastSuccessLeaderSyncTime = syncLeaderTask.getEndTime();
            }
        }

        int countSyncSuccess(Collection<String> sessions) {
            int count = 0;
            for (String session : sessions) {
                MigratingTask t = migratingTasks.get(session);
                // not contains forceSuccess
                if (t != null && t.task.isSuccess()) {
                    count++;
                }
            }
            return count;
        }

        boolean isAnywaySuccess(Collection<String> sessions) {
            if (sessions.isEmpty()) {
                return false;
            }
            // contains forceSuccess
            for (String session : sessions) {
                final MigratingTask t = migratingTasks.get(session);
                if (t == null) {
                    return false;
                }
                // contains forceSuccess
                if (t.forceSuccess) {
                    continue;
                }
                if (!t.task.isSuccess()) {
                    // could not use isFailed to replace !isSuccess
                    return false;
                }
            }
            return true;
        }

        @Override
        public boolean isMigrated() {
            return this.migrated;
        }

        @Override
        public Slot getSlot() {
            return this.slot;
        }
    }

    static class MigratingTask {
        final long createTimestamp = System.currentTimeMillis();
        final String sessionIp;
        KeyedTask<SyncSessionTask> task;
        int tryCount;
        boolean forceSuccess;

        MigratingTask(String sessionIp, KeyedTask<SyncSessionTask> task) {
            this.sessionIp = sessionIp;
            this.task = task;
        }
    }

    private static final class SyncSessionTask implements Runnable {
        final long startTimestamp = System.currentTimeMillis();
        final String syncDataCenter;
        final boolean migrating;
        final long slotsEpoch;
        final Slot slot;
        final String sessionIp;
        final SlotDiffSyncer syncer;
        final SessionNodeExchanger sessionNodeExchanger;
        final SyncContinues continues;
        final Map<String, DatumSummary> summary;

        SyncSessionTask(String syncDataCenter, boolean migrating, long slotsEpoch, Slot slot, String sessionIp, SlotDiffSyncer syncer, SessionNodeExchanger sessionNodeExchanger, SyncContinues continues, Map<String, DatumSummary> summary) {
            this.syncDataCenter = syncDataCenter;
            this.migrating = migrating;
            this.slotsEpoch = slotsEpoch;
            this.slot = slot;
            this.sessionIp = sessionIp;
            this.syncer = syncer;
            this.sessionNodeExchanger = sessionNodeExchanger;
            this.continues = continues;
            this.summary = summary;
        }

        public void run() {
            boolean success = false;
            try {

                success = syncer.syncSession(syncDataCenter, slot.getId(), sessionIp, slot.getLeaderEpoch(), sessionNodeExchanger, slotsEpoch, continues, summary);
                if (!success) {
                    // sync failed
                    throw new RuntimeException("sync session failed");
                }
            } catch (Throwable e) {
                if (migrating) {
                    observeLeaderMigratingFail(slot.getId(), sessionIp);
                    logger.error("[migrating]failed: {}, slot={}", sessionIp, slot.getId(), e);
                } else {
                    logger.error("[syncSession]failed: {}, slot={}", sessionIp, slot.getId(), e);
                }
                // rethrow silence exception, notify the task is failed
                throw TaskErrorSilenceException.INSTANCE;
            } finally {
                logger.info("{},{},{},{},span={}", success ? 'Y' : 'N', migrating ? 'M' : 'S', slot.getId(), sessionIp, System.currentTimeMillis() - startTimestamp);
            }
        }


    }

    private final class SyncingWatchDog extends WakeUpLoopExecuteTask {

        @Override
        public void _execute_() {
            try {
                processUpdating();
                syncWatch();
            } catch (Throwable e) {
                logger.error("[syncWatch]failed to do sync watching", e);
            }
        }

        @Override
        public int getWaitingMillis() {
            return 200;
        }
    }
}
